package com.bestcem.xm.member.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.bestcem.xm.common.core.domain.web.ServiceResult;
import com.bestcem.xm.common.core.uitls.CommonUtil;
import com.bestcem.xm.common.core.uitls.DateUtil;
import com.bestcem.xm.common.core.uitls.IdUtil;
import com.bestcem.xm.member.config.PerformanceConfig;
import com.bestcem.xm.member.dao.MemberDao;
import com.bestcem.xm.member.dao.SyncRuleDao;
import com.bestcem.xm.member.entity.pojo.MemberDO;
import com.bestcem.xm.member.entity.pojo.SyncRuleDO;
import com.bestcem.xm.member.enums.*;
import com.bestcem.xm.member.grpc.client.PushClient;
import com.bestcem.xm.member.mq.message.MemberMessage;
import com.bestcem.xm.member.mq.message.SyncWechatAccountInfoMessage;
import com.bestcem.xm.member.mq.message.WechatMemberSyncMessage;
import com.bestcem.xm.member.mq.message.WechatOfficialAccountGrantedMessage;
import com.bestcem.xm.member.service.*;
import com.bestcem.xm.member.service.cache.CacheMemberService;
import com.bestcem.xm.member.service.dto.*;
import com.bestcem.xm.member.service.mq.send.MemberMessageSendService;
import com.bestcem.xm.member.service.param.AddFootprintParam;
import com.bestcem.xm.member.util.MemberUtil;
import com.google.common.collect.Iterables;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.bestcem.xm.member.constant.Constants.FIVE_MIN_UNIT_MILLIS;

/**
 * 微信同步联系人相关
 *
 * @author yf.wang <yf.wang@idiaoyan.com>
 * @version V1.0
 * @date 2022/3/29
 */

@Slf4j
@Service
public class WechatServiceImpl implements WechatService {

    private static final Logger EMAIL_LOG = LoggerFactory.getLogger("EmailErrorLog");

    private static final Logger MEMBER_PERFORMANCE_LOG = LoggerFactory.getLogger("MemberPerformanceLog");

    @Resource
    private PushClient pushClient;
//
//    @Resource
//    private RabbitMessageSender rabbitMessageSender;
//
//    @Resource
//    private RabbitMqConstantConfig rabbitMqConstantConfig;

    @Resource
    private MemberDao memberDao;

    @Resource
    private SequenceService sequenceService;

    @Resource
    private GroupService groupService;

    @Resource
    private MemberUtil memberUtil;

    @Resource
    private FootprintService footprintService;

    @Resource
    private SourceCardService sourceCardService;

    @Resource
    private SyncRuleDao syncRuleDao;

    @Resource
    private MemberMessageSendService memberMessageSendService;

    @Resource
    private CacheMemberService cacheMemberService;

    private final static Integer RETRY_COUNT = 3;

    @Override
    public void sendWxSyncMessage(WechatMemberSyncMessage message) {
//        try {
//            rabbitMessageSender.sendMsg(rabbitMqConstantConfig.getAppBpMemberMemberWechatMemberSyncExchange(),
//                    rabbitMqConstantConfig.getAppBpMemberMemberWechatMemberSyncQueue(), MEMBER_WECHAT_MEMBER_SYNC_TOPIC, message);
//        } catch (Exception e) {
//            EMAIL_LOG.error("【联系人】 发布微信同步信息失败", e);
//        }

        //stram mq 2022.05.12
        memberMessageSendService.memberWechatMemberSyncSend(message);
    }

    @Override
    public ServiceResult<Void> sendWxSyncMessage(String orgId) {
        // 获取已授权的公众号列表
        ServiceResult<List<WechatOfficialAccountInfoDTO>> listServiceResult =
                pushClient.listAccountInfo(orgId);
        // 循环发送消息
        if (listServiceResult.isSuccess()) {
            List<WechatOfficialAccountInfoDTO> data = listServiceResult.getData();
            if (CollUtil.isNotEmpty(data)) {
                if (log.isInfoEnabled()) {
                    log.info("该 orgId{} 已授权的公众号 {} 个", orgId, data.size());
                }
                data.forEach(item -> {
                    String appId = item.getAppId();
                    Integer version = item.getVersion();
                    WechatMemberSyncMessage message = new WechatMemberSyncMessage();
                    message.setOrgId(orgId);
                    message.setVersion(version);
                    message.setAppId(appId);
                    message.setRetryCount(0);
                    sendWxSyncMessage(message);
                });
            }
            return ServiceResult.success();
        } else {
            log.error("{}", listServiceResult.getMsg());
            return ServiceResult.fail(listServiceResult.getMsg());
        }
    }

    @Override
    public void scheduleSyncWxMember(WechatMemberSyncMessage message) {
        try {
            handleSyncWxMember(message.getOrgId(), message.getAppId(), message.getVersion(), message.getBatchId());
        } catch (Exception e) {
            Integer retryCount = message.getRetryCount();
            if (retryCount >= RETRY_COUNT) {
                // 不在发送消息
                EMAIL_LOG.error("【联系人】【定时任务】处理微信公众号粉丝同步联系人{}, 重试次数超过 {} 次", JSONObject.toJSONString(message), RETRY_COUNT, e);
            } else {
                // 发送5min延迟消息
                message.setRetryCount(retryCount + 1);
                log.error("【联系人】【定时任务】处理微信公众号粉丝同步联系人失败，开始重试 {}", message.getRetryCount(), e);
                Date delayDate = new Date(System.currentTimeMillis() + FIVE_MIN_UNIT_MILLIS);
//                rabbitMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberWechatMemberScheduleSyncDelayExchange(),
//                        rabbitMqConstantConfig.getAppMemberWechatMemberScheduleSyncDelayRoutingKey(), MemberTopics.MEMBER_WECHAT_MEMBER_SCHEDULE_SYNC_DELAY_TOPIC, message, delayDate);
                memberMessageSendService.wechatMemberScheduleSyncDelay(message, delayDate);
            }
        }
    }

    @Override
    public void syncWxMember(WechatMemberSyncMessage message) {
        try {
            handleSyncWxMember(message.getOrgId(), message.getAppId(), message.getVersion(), message.getBatchId());
        } catch (Exception e) {
            Integer retryCount = message.getRetryCount();
            if (retryCount >= RETRY_COUNT) {
                // 不在发送消息
                EMAIL_LOG.error("【联系人】处理微信公众号粉丝同步联系人{}, 重试次数超过 {} 次", JSONObject.toJSONString(message), RETRY_COUNT, e);
            } else {
                // 发送5min延迟消息
                message.setRetryCount(retryCount + 1);
                log.error("【联系人】处理微信公众号粉丝同步联系人失败，开始重试 {}", message.getRetryCount(), e);
                Date delayDate = new Date(System.currentTimeMillis() + FIVE_MIN_UNIT_MILLIS);
//                rabbitMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberWechatMemberSyncDelayExchange(),
//                        rabbitMqConstantConfig.getAppMemberWechatMemberSyncDelayRoutingKey(), MemberTopics.MEMBER_WECHAT_MEMBER_SYNC_DELAY_TOPIC, message, delayDate);
                memberMessageSendService.wechatMemberSyncDelayDelaySend(message, delayDate);
            }
        }
    }

    @Override
    public void syncWechatAccountInfo(SyncWechatAccountInfoMessage message) {
        String batchId = message.getBatchId();
        // 获取该联系人
        if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
            MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】获取联系人开始  batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
        }
        MemberDO memberDO = memberDao.findOneByOrgIdAndMemberTypeAndStatusAndAppIdAndOpenId(message.getOrgId(), MemberTypeEnum.INDIVIDUAL,
                MemberStatusEnum.ACTIVATED, message.getAppId(), message.getOpenId());
        if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
            MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】获取联系人结束 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
        }
        if (ObjectUtil.isEmpty(memberDO)) {
            memberDO = new MemberDO();
            // orgId
            memberDO.setOrgId(message.getOrgId());
            memberDO.setMemberType(MemberTypeEnum.INDIVIDUAL.getIndex());
            memberDO.setStatus(MemberStatusEnum.ACTIVATED.getIndex());
            // openId
            memberDO.setOpenId(message.getOpenId());
            // appId
            memberDO.setAppId(message.getAppId());
            // unionId
            memberDO.setUnionId(message.getUnionId());
            saveMember(memberDO, message.getOrgId(), message.getOrgCode(), message.getOrgName(), batchId);
        } else {
            // 更新unionId
            memberDO.setUnionId(message.getUnionId());
            updateMember(memberDO, message.getOrgCode(), message.getOrgName(), batchId);
        }
    }

    @Override
    public void publishSyncMemberSlices() {
        //  获取所有授权的orgId
        int page = 1;
        int size = 1000;
        List<AuthorizedOrgInfoDTO> authorizedOrgInfoDTOS = null;
        do {
            // 获取 授权的orgId 列表
            if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】获取授权的orgId列表开始 timestamp: {}", System.currentTimeMillis());
            }
            ServiceResult<List<AuthorizedOrgInfoDTO>> listServiceResult =
                    pushClient.listAuthorizedOrgInfo(page, size);
            if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】获取授权的orgId列表结束  timestamp: {}", System.currentTimeMillis());
            }
            if (listServiceResult.isSuccess()) {
                authorizedOrgInfoDTOS = listServiceResult.getData();
                if (log.isInfoEnabled()) {
                    log.info("第 {} 批 获取 {} 组织", page, authorizedOrgInfoDTOS.size());
                }
                if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                    MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】第 {} 批 获取 {} 组织", page, authorizedOrgInfoDTOS.size());
                }
                if (CollUtil.isNotEmpty(authorizedOrgInfoDTOS)) {
                    // 组装 orgId列表
                    List<String> orgIds = authorizedOrgInfoDTOS.stream()
                            .map(AuthorizedOrgInfoDTO::getId).collect(Collectors.toList());
                    // 组装 map
                    Map<String, List<AuthorizedOrgInfoDTO>> authorizedOrgInfoMap =
                            authorizedOrgInfoDTOS.stream().collect(Collectors.groupingBy(AuthorizedOrgInfoDTO::getId));
                    // 判断是否开始微信同步
                    if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                        MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】判断是否开始微信同步开始 timestamp: {}", System.currentTimeMillis());
                    }
                    List<String> syncOrgIds = syncRuleDao.findByOrgIdsAndMemberTypeAndWechatSyncEnabled(orgIds, MemberTypeEnum.INDIVIDUAL.getIndex(), SyncTypeEnum.NOW.getIndex());
                    if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                        MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】判断是否开始微信同步结束 timestamp: {}", System.currentTimeMillis());
                    }
                    if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                        MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】第 {} 批 共有 {} 组织开启微信同步", page, syncOrgIds.size());
                    }
                    if (CollUtil.isNotEmpty(syncOrgIds)) {
                        syncOrgIds.forEach(item -> {
                            // 发送消息
                            if (CollUtil.isNotEmpty(authorizedOrgInfoMap)) {
                                if (authorizedOrgInfoMap.containsKey(item)) {
                                    AuthorizedOrgInfoDTO authorizedOrgInfoDTO = authorizedOrgInfoMap.get(item).get(0);

                                    WechatMemberSyncMessage message = new WechatMemberSyncMessage();
                                    message.setRetryCount(0);
                                    message.setOrgId(item);
                                    message.setAppId(authorizedOrgInfoDTO.getAppId());
                                    message.setVersion(authorizedOrgInfoDTO.getVersion());
                                    sendWxScheduleSyncMessage(message);
                                }
                            }
                        });
                    }
                    page++;
                }
            }
        } while (CollUtil.isNotEmpty(authorizedOrgInfoDTOS) && authorizedOrgInfoDTOS.size() >= size);
    }

    @Override
    public void sendWxScheduleSyncMessage(WechatMemberSyncMessage message) {
//        try {
//            message.setBatchId(UUIDUtil.uuid());
//            rabbitMessageSender.sendMsg(rabbitMqConstantConfig.getAppBpMemberWechatMemberScheduleSyncExchange(),
//                    rabbitMqConstantConfig.getAppBpMemberWechatMemberScheduleSyncQueue(), MEMBER_WECHAT_MEMBER_SCHEDULE_SYNC_TOPIC, message);
//        } catch (Exception e) {
//            EMAIL_LOG.error("【联系人】 发布微信同步信息失败", e);
//        }
        //stream mq /2022.05.10
        memberMessageSendService.memberWechatMemberScheduleSend(message);

    }

    private String handleAddress(String country, String province, String city) {
        List<String> address = new ArrayList<>();
        address.add(StrUtil.isBlank(country) ? "" : country);
        address.add(StrUtil.isBlank(province) ? "" : province);
        address.add(StrUtil.isBlank(city) ? "" : city);
        address.add("");
        return JSONObject.toJSONString(address);
    }

    @Override
    public void syncWechatOfficialAccountGranted(WechatOfficialAccountGrantedMessage message) {
        try {
            // 判断该组织是否开启微信同步
            SyncRuleDO syncRuleDO = syncRuleDao.getByOrderIdAndMemberType(message.getOrgId(), MemberTypeEnum.INDIVIDUAL.getIndex());
            if (syncRuleDO != null && syncRuleDO.getWechatSyncEnabled() == 0) {
                String batchId = IdUtil.uuid();
                handleSyncWxMember(message.getOrgId(), message.getAppId(), message.getVersion(), batchId);
            } else {
                log.error("该组织 {} 没有开启公众号授权", message.getOrgId());
            }
        } catch (Exception e) {
            // 发送延迟消息
            WechatMemberSyncMessage errorMessage = new WechatMemberSyncMessage();
            errorMessage.setRetryCount(1);
            errorMessage.setAppId(message.getAppId());
            errorMessage.setOrgId(message.getOrgId());
            errorMessage.setVersion(message.getVersion());
            // 发送5min延迟消息
            Date delayDate = new Date(System.currentTimeMillis() + FIVE_MIN_UNIT_MILLIS);
//            rabbitMessageSender.sendDelayMsg(rabbitMqConstantConfig.getAppMemberDelayExchange(),
//                    rabbitMqConstantConfig.getAppMemberDelayRoutingKey(), MemberTopics.MEMBER_WECHAT_MEMBER_SYNC_DELAY_TOPIC, message, delayDate);

            memberMessageSendService.wechatMemberSyncDelaySend(errorMessage, delayDate);

        }
    }

    /**
     * 获取 公众号粉丝openId 列表
     *
     * @param orgId
     * @param appId
     * @param version
     * @param nextOpenId
     * @return
     */
    private WechatOpenIdDTO getWxInfo(String orgId, String appId, Integer version, String nextOpenId) {
        ServiceResult<WechatOpenIdDTO> result =
                pushClient.listOpenIds(orgId, appId, version, nextOpenId);
        if (result.isSuccess()) {
            WechatOpenIdDTO data = result.getData();
            return data;
        } else {
            log.error("listOpenIds 失败 {}", result.getMsg());
        }
        return null;
    }


    /**
     * 新增联系人
     *
     * @param member
     * @param orgId
     * @param orgName
     * @param orgCode
     * @param batchId
     */
    private void saveMember(MemberDO member, String orgId, String orgName, String orgCode, String batchId) {
        // 联系人限额判断

        ServiceResult<String> result = memberUtil.checkMemberNumber(orgId);
        if (!result.isSuccess()) {
            log.error("该组织 {} 个人联系人 {} ", orgId, result.getMsg());
            return;
        }

        String groupSyncToken = CommonUtil.getUUID();
        member.setGroupSyncToken(groupSyncToken);
        member.setMemberNo(getMemberNo(orgId));
        member.setSource("data");
        member.setCreateTime(DateUtil.getCommonDate());
        member.setUpdateTime(member.getCreateTime());
        if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
            MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】保存联系人开始 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
        }
        String id = memberDao.wechatSaveMember(member);
        if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
            MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】保存联系人结束 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
        }
        // 分组同步令牌
        groupService.sendMemberCreationUpdMessage(orgId, id, groupSyncToken);

        // 推送bi
        // 推送创建的联系人
        MemberMessage message = MemberUtil.memberDoToMemberMessage(member);
        message.setOrgCode(orgCode);
        message.setOrgName(orgName);
        memberUtil.publishCreateMember(message);

        // 推送成为联系人足迹点
        JSONObject footprint = new JSONObject();
        footprint.put("title", "成为联系人");
        AddFootprintParam addFootprintParam = new AddFootprintParam();
        addFootprintParam.setOrgId(orgId);
        addFootprintParam.setMemberId(member.getId());
        addFootprintParam.setType(FootprintTypeEnum.MEMBER.getIndex());
        addFootprintParam.setSourceKey(member.getId());
        addFootprintParam.setContent(footprint);
        addFootprintParam.setFootprintTime(member.getCreateTime());
        footprintService.publishFootprint(addFootprintParam);

        // 推送联系人名片
        sourceCardService.publishSourceCard(member, CardSourceTypeEnum.DATA.getIndex());
    }

    /**
     * 更新联系人
     *
     * @param member
     * @param orgName
     * @param orgCode
     * @param batchId
     */
    private void updateMember(MemberDO member, String orgName, String orgCode, String batchId) {
        String groupSyncToken = CommonUtil.getUUID();
        member.setGroupSyncToken(groupSyncToken);
        member.setUpdateTime(DateUtil.getCommonDate());
        if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
            MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】更新联系人开始 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
        }
        memberDao.updateMember(member, true);
        if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
            MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】更新联系人结束 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
        }
        // 分组同步令牌
        groupService.sendMemberCreationUpdMessage(member.getOrgId(), member.getId(), groupSyncToken);

        // 推送bi
        // 推送更新的联系人
        MemberMessage message = MemberUtil.memberDoToMemberMessage(member);
        message.setOrgCode(orgCode);
        message.setOrgName(orgName);
        memberUtil.publishUpdateMember(message);

    }

    private String getMemberNo(String orgId) {
        ServiceResult<Integer> sequenceResult =
                sequenceService.getSequence("member:memberNo_INDIVIDUAL_" + orgId, 1L);
        String memberNo = "C" + NumberUtil.decimalFormat("00000000", sequenceResult.getData());
        if (log.isInfoEnabled()) {
            log.info("该联系人编号 {}", memberNo);
        }
        return memberNo;
    }

    /**
     * 从本地缓存获取 减少与python的交互
     * 若缓存数据不存在 则请求 grpc
     *
     * @param orgId
     * @return
     */
    private OrganizationDTO getOrganization(String orgId) {
        return cacheMemberService.getOrganizationDTO(orgId);
    }

    /**
     * 粉丝 同步到 联系人
     *
     * @param orgId
     * @param appId
     * @param version
     * @param batchId
     */
    private void handleSyncWxMember(String orgId, String appId, Integer version, String batchId) {
        // 获取 该公众号所有粉丝的openId
        String nextOpenId = null;
        do {
            // 获取 公众号粉丝openId 列表
            if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】获取公众号粉丝openId列表开始 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
            }
            // 获取 公众号粉丝openId 列表
            WechatOpenIdDTO wechatOpenIdDTO =
                    getWxInfo(orgId, appId, version, nextOpenId);
            if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】获取公众号粉丝openId列表结束 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
            }
            List<String> openIds = new ArrayList<>();
            if (ObjectUtil.isNotEmpty(wechatOpenIdDTO)) {
                // nextOpenId 若存在 则表示有下一页数据
                nextOpenId = wechatOpenIdDTO.getNextOpenId();
                openIds.addAll(wechatOpenIdDTO.getOpenIds());
            }
            if (CollUtil.isNotEmpty(openIds)) {
                if (log.isInfoEnabled()) {
                    log.info("该组织 {} 公众号 {} 粉丝数量 {}", orgId, appId, openIds.size());
                }
                OrganizationDTO organization = getOrganization(orgId);
                // 分批获取详细数据
                // 微信那边限制 最多批量获取 100
                Iterables.partition(openIds, 100).forEach(items -> {
                    // 根据 openId 获取该粉丝的unionId
                    if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                        MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】根据openId获取该粉丝的unionId开始 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
                    }
                    ServiceResult<List<WechatUserBasicInfoDTO>> listUserBasicInfoResult =
                            pushClient.listUserBasicInfo(items, appId);
                    if (MEMBER_PERFORMANCE_LOG.isInfoEnabled() && PerformanceConfig.WECHAT_SYNC_MEMBER_LOG_ENABLED) {
                        MEMBER_PERFORMANCE_LOG.info("【微信同步联系人性能监控】根据openId获取该粉丝的unionId结束 batchId: {} timestamp: {}", batchId, System.currentTimeMillis());
                    }
                    if (listUserBasicInfoResult.isSuccess()) {
                        List<WechatUserBasicInfoDTO> data = listUserBasicInfoResult.getData();
                        if (log.isInfoEnabled()) {
                            log.info("该组织 {} 公众号 {} 粉丝数据 {}", orgId, appId, data.size());
                        }
                        if (CollUtil.isNotEmpty(data)) {
                            data.forEach(item -> {
                                SyncWechatAccountInfoMessage message = new SyncWechatAccountInfoMessage();
                                message.setAppId(item.getAppId());
                                message.setOpenId(item.getOpenId());
                                message.setUnionId(item.getUnionId());
                                message.setOrgId(orgId);
                                message.setOrgCode(organization.getCode());
                                message.setOrgName(organization.getName());
                                message.setBatchId(IdUtil.uuid());
//                                try {
//                                    rabbitMessageSender.sendMsg(rabbitMqConstantConfig.getAppMemberSyncWechatAccountInfoExchange(),
//                                            rabbitMqConstantConfig.getAppMemberSyncWechatAccountInfoQueue(), MEMBER_SYNC_WECHAT_MEMBER_ACCOUNT_INFO_TOPIC, message);
//                                } catch (Exception e) {
//                                    EMAIL_LOG.error("【联系人】 发布微信同步信息失败", e);
//                                }
                                //stream mq /2022.05.12
                                memberMessageSendService.syncWechatAccountInfoSend(message);
                            });
                        }
                    }
                });
            }
        } while (StrUtil.isNotBlank(nextOpenId));
    }
}
